package ink.feta.task.runner;

import ink.feta.task.job.registry.JobAPIProxy;
import ink.feta.task.job.registry.JobRepository;
import ink.feta.task.service.ElasticJobService;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
import org.apache.shardingsphere.elasticjob.infra.concurrent.BlockUtils;
import org.apache.shardingsphere.elasticjob.lite.internal.config.ConfigurationNode;
import org.apache.shardingsphere.elasticjob.lite.internal.config.ConfigurationService;
import org.apache.shardingsphere.elasticjob.lite.internal.election.LeaderService;
import org.apache.shardingsphere.elasticjob.lite.internal.failover.FailoverService;
import org.apache.shardingsphere.elasticjob.lite.internal.guarantee.GuaranteeService;
import org.apache.shardingsphere.elasticjob.lite.internal.instance.InstanceService;
import org.apache.shardingsphere.elasticjob.lite.internal.listener.AbstractJobListener;
import org.apache.shardingsphere.elasticjob.lite.internal.schedule.JobRegistry;
import org.apache.shardingsphere.elasticjob.lite.internal.sharding.ExecutionService;
import org.apache.shardingsphere.elasticjob.lite.internal.storage.JobNodePath;
import org.apache.shardingsphere.elasticjob.lite.internal.storage.JobNodeStorage;
import org.apache.shardingsphere.elasticjob.lite.internal.trigger.TriggerService;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.Optional;

@Component
public class ZookeeperListenRunner implements ApplicationRunner {
    private static Logger logger = LoggerFactory.getLogger(ZookeeperListenRunner.class);

    @Resource
    CoordinatorRegistryCenter coordinatorRegistryCenter;

    @Resource
    ElasticJobService elasticJobService;

    @Resource
    JobAPIProxy jobAPIProxy;


    private JobRegistry jobRegistry = JobRegistry.getInstance();
    private JobRepository jobRepository = JobRepository.getInstance();

    @Override
    public void run(ApplicationArguments args) throws Exception {
        logger.info("添加Listener ");
        CuratorFramework curatorFramework = (CuratorFramework) coordinatorRegistryCenter.getRawClient();
        CuratorCache curatorCache = CuratorCache.build(curatorFramework, "/");
        CuratorCacheListener curatorCacheListener = CuratorCacheListener.builder().forAll(new AbstractJobListener() {
            @Override
            protected void dataChanged(String path, Type type, String data) {
                switch (type) {
                    case NODE_CREATED:
                        nodeCreated(path, type, data);
                        break;
                    case NODE_DELETED:
                        nodeDeleted(path, type, data);
                        break;
                    case NODE_CHANGED:
                        nodeChanged(path, type, data);
                        break;
                }
            }
        }).build();
        curatorCache.listenable().addListener(curatorCacheListener);
        curatorCache.start();
        logger.info("添加Listener 完成");
    }

    private void nodeCreated(String path, CuratorCacheListener.Type type, String data) {
        if (logger.isDebugEnabled()) {
            logger.debug(" path: [{}], type: [{}], data: [{}]", path, type, data);
        }
        Optional<String> jobName = getJobName(path);
        if (jobName.isPresent()) {
            String job = jobName.get();

            ConfigurationNode configurationNode = new ConfigurationNode(job);
            if (configurationNode.isConfigPath(path)) {
                logger.info("获得job [{}] ", job);
                BlockUtils.waitingShortTime();
                if (!jobRegistry.isJobRunning(job)) {
                    ConfigurationService configurationService = new ConfigurationService(coordinatorRegistryCenter, job);
                    JobConfiguration configuration = configurationService.load(false);
                    elasticJobService.createJob(configuration);
                } else {
                    logger.info("JOB: [{}] is running", job);
                }
            } else if (path.equals("/" + job + "/delete")) {
                logger.info("任务 [{}] 正在删除", job);

                FailoverService failoverService = new FailoverService(coordinatorRegistryCenter, job);
                failoverService.removeFailoverInfo();

                ExecutionService executionService = new ExecutionService(coordinatorRegistryCenter, job);
                executionService.clearAllRunningInfo();

                TriggerService triggerService = new TriggerService(coordinatorRegistryCenter, job);
                triggerService.removeTriggerFlag();

                GuaranteeService guaranteeService = new GuaranteeService(coordinatorRegistryCenter, job);
                guaranteeService.clearAllStartedInfo();
                guaranteeService.clearAllCompletedInfo();

                InstanceService instanceService = new InstanceService(coordinatorRegistryCenter, job);
                instanceService.removeInstance();

                jobRepository.getJobScheduler(job).shutdown();

                LeaderService leaderService = new LeaderService(coordinatorRegistryCenter, job);
                leaderService.removeLeader();

                jobRegistry.shutdown(job);

                logger.info("任务 [{}] 删除 完成", job);

            }
        }
    }

    private void nodeChanged(String path, CuratorCacheListener.Type type, String data) {
        if (logger.isDebugEnabled()) {
            logger.debug(" path: [{}], type: [{}], data: [{}]", path, type, data);
        }
    }

    private void nodeDeleted(String path, CuratorCacheListener.Type type, String data) {
        if (logger.isDebugEnabled()) {
            logger.debug(" path: [{}], type: [{}], data: [{}]", path, type, data);
        }
    }

    private Optional<String> getJobName(String path) {
        String[] split = path.replace("/", " ").trim().split(" ");
        return Arrays.stream(split).findFirst();
    }
}
